package com.example.rpc_core.register.impl;

import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.example.rpc_core.cache.RegisterServiceMultiCache;
import com.example.rpc_core.config.RegistryConfig;
import com.example.rpc_core.model.ServiceInfo;
import com.example.rpc_core.register.Register;
import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
@Slf4j
public class EtcdRegister implements Register {
    private Client client;
    private KV KVClient;
    /**
     * 保存续约id的集合
     */
    private final ConcurrentMap<String, Long> keyToLeaseIdMap = new ConcurrentHashMap<>();
    /**
     * 定义根节点
     *
     * @param registryConfig
     */
    private static final String ROOT = "/rpc/";
//    /**
//     * 注册中心服务缓存（这是只支持单服务）
//     */
//    private final RegisterServiceCache registerServiceCache=new RegisterServiceCache();

    /**
     * 注册中心服务缓存（支持多服务）
     */
    private final RegisterServiceMultiCache registerServiceMultiCache=new RegisterServiceMultiCache();

    /**
     * 正在监听的 key 集合
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();

    /**
     * 心跳检测集合
     */
    private final Set<String> localRegisterServiceKeySet = new HashSet<>();


    /**
     * 初始化
     *
     * @param registryConfig
     */
    @Override
    public void init(RegistryConfig registryConfig) {
        client = Client.builder().endpoints(registryConfig.getRegistryAddress())
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout())).build();
        KVClient = client.getKVClient();
        //启动心跳检测
        heartbeat();
    }


    @Override
    public void register(ServiceInfo serviceInfo) throws ExecutionException, InterruptedException {
        //创建一个新的客户端
        Lease leaseClient = client.getLeaseClient();
        //创建租约(时间为40秒)
        long id = leaseClient.grant(30).get().getID();
        //设置键值对
        String key = ROOT + serviceInfo.getServiceNodeKey();
        //储存键值对
        KVClient.put(
                ByteSequence.from(key, StandardCharsets.UTF_8),
                ByteSequence.from(JSONUtil.toJsonStr(serviceInfo), StandardCharsets.UTF_8),
                PutOption.newBuilder().withLeaseId(id).build()).get();
        //将Lease ID保存
        keyToLeaseIdMap.put(key, id);
        //存储到心跳检测集合
        localRegisterServiceKeySet.add(key);

    }

    @Override
    public void unRegister(ServiceInfo serviceInfo) {
        String key = ROOT + serviceInfo.getServiceNodeKey();
        KVClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8));
        localRegisterServiceKeySet.remove(key);
    }

    @Override
    public List<ServiceInfo> serviceDiscovery(String serviceKey) {

        //优先从缓存中获取服务
//        List<ServiceInfo> serviceInfos = registerServiceCache.readCache();
        List<ServiceInfo> serviceInfos = registerServiceMultiCache.readCache(serviceKey);
        if (serviceInfos != null) {
            return serviceInfos;
        }

        String prefix = ROOT + serviceKey;
        try {
            //前缀查询
            //配置获取规制
            GetOption build = GetOption.newBuilder().isPrefix(true).build();
            List<KeyValue> kvs = KVClient
                    .get(ByteSequence.from(prefix, StandardCharsets.UTF_8), build)
                    .get()
                    .getKvs();
            List<ServiceInfo> collect = kvs.stream()
                    .map(kv -> {
                        //监听对应的key
                        String key= kv.getKey().toString(StandardCharsets.UTF_8);
                        watch(key);
                        //将对应的value序列化返回
                        String value = kv.getValue().toString(StandardCharsets.UTF_8);
                        return JSONUtil.toBean(value, ServiceInfo.class);
                    }).collect(Collectors.toList());
//            registerServiceCache.writeCache(collect);
            registerServiceMultiCache.writeCache(serviceKey,collect);
            return collect;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void destroy() {
        System.out.println("销毁当前节点");
        for (String key:localRegisterServiceKeySet
             ) {
            try {
                KVClient.delete(ByteSequence.from(key,StandardCharsets.UTF_8))
                        .get();
            } catch (Exception e) {
                throw new RuntimeException(key+"节点下线失败");
            }
        }
        if (client != null) {
            client.close();
        }
        if (KVClient != null) {
            KVClient.close();
        }

    }

    /**
     * 心跳检测实现
     */
    @Override
    public void heartbeat() {
        // 启动定时任务
        if (CronUtil.getScheduler().isStarted()) {
            CronUtil.stop();
        }
        // 配置秒级调度（确保精确到秒）
        CronUtil.setMatchSecond(true);
        //10秒一次续约
        CronUtil.schedule(("*/10 * * * * ?"), new Task() {
            @Override
            public void execute() {
                for (String key:localRegisterServiceKeySet
                     ) {
                    try {
                        // 获取leaseId
                        Long leaseId = keyToLeaseIdMap.get(key);
                        if (leaseId == null) {
                            // 无 leaseId 则重新注册
                            reRegister(key);
                            continue;
                        }
                        // 续约
                        client.getLeaseClient().keepAliveOnce(leaseId);
                        log.debug("续约成功: {}", key);
                    } catch (Exception e) {
                        throw new RuntimeException(key+"续签失败");
                    }
                }
            }
        });
        //15分钟一次强制刷新
        CronUtil.schedule("5 0/15 * * * * ?", new Task() {
            @Override
            public void execute() {
                for (String key : localRegisterServiceKeySet) {
                    try {
                        reRegister(key); // 强制刷新
                        log.info("定时刷新完成: {}", key);
                    } catch (Exception e) {
                        log.error("定时刷新失败: {}", key, e);
                    }
                }
            }
        });
        CronUtil.start();
    }
    // 重新注册逻辑（私有方法）
    private void reRegister(String key) {
        try {
            GetResponse resp = KVClient.get(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            if (!resp.getKvs().isEmpty()) {
                String value = resp.getKvs().get(0).getValue().toString(StandardCharsets.UTF_8);
                ServiceInfo info = JSONUtil.toBean(value, ServiceInfo.class);
                register(info); // 调用原有注册逻辑
            }
        } catch (Exception e) {
            log.error("重新注册失败: {}", key, e);
        }
    }
    /**
     * 监听的实现
     * @param serviceKey
     */
    @Override
    public void watch(String serviceKey) {
        Watch watchClient = client.getWatchClient();
        boolean add = watchingKeySet.add(serviceKey);
        if(add){
           watchClient.watch(ByteSequence.from(serviceKey,StandardCharsets.UTF_8),watchResponse ->{
               for (WatchEvent event:watchResponse.getEvents()) {
                   switch (event.getEventType()) {
                       //当删除时触发
                       case DELETE:
                           registerServiceMultiCache.clearCache(serviceKey);
                           break;
                       case PUT:
                       default:
                           break;
                   }
               }
        });

    }
    }
}